package com.zzr.demo.redis.II.service.impl;


import com.zzr.demo.redis.II.service.MessageQueueService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @author: xiaomi
 * @date: 2021/10/11
 * @description: 由于2.1x 版本没有实装 redis stream ,单单升级 spring-data 会有一系列的问题，于是考虑换个 服务。
 */
@Service
@Slf4j
public class MessageQueueByStreamServiceImpl implements MessageQueueService {

    final String MQ_STREAM_KEY = "mq:stream";
    @Autowired
    StringRedisTemplate stringRedisTemplate;

    /**
     * @param key
     * @param value
     * @return recordId
     */
    @Override
    public String addMessage(String key, String value) {
//        throw new NotImplementedException();
//        ObjectRecord<String, String> record = StreamRecords.newRecord()
//                .in("my-stream")
//                .ofObject("my-value");
        //客户端生成ID
        StringRecord stringRecord = StreamRecords.string(Collections.singletonMap(key, value)).withStreamKey(MQ_STREAM_KEY);
        log.info("steam k:{},v:{}", key, value);
        RecordId recordId = stringRedisTemplate.opsForStream().add(stringRecord);
        log.info("add recordId:{}", recordId);
        // 是否是自动生成的
        boolean autoGenerated = recordId.shouldBeAutoGenerated();
        // id值
        String v = recordId.getValue();
        // 序列号部分
        long sequence = recordId.getSequence();
        // 时间戳部分
        long timestamp = recordId.getTimestamp();
        log.info("是否是自动生成的:{},recordId:{},sequence:{},timestamp:{}", autoGenerated, v, sequence, timestamp);
        return v;
    }

    @Override
    public String getMessage(String recordId) {
        //StreamReadOptions.empty().count()
        //StreamOffset.latest(MQ_STREAM_KEY) 使用 $ 可以用于阻塞获取中
        List<MapRecord<String, Object, Object>> messages = stringRedisTemplate.opsForStream().read(StreamOffset.fromStart(MQ_STREAM_KEY));
        String msg = "";
        for (MapRecord<String, Object, Object> message : messages) {
            log.info("RecordId:{},Stream:{},Value:{}", message.getId(), message.getStream(), message.getValue());
            msg += message.getValue().toString();
        }
        return msg;

    }

    /**
     * 返回队列的长度
     *
     * @return
     */
    @Override
    public long size() {
        long size = stringRedisTemplate.opsForStream().size(MQ_STREAM_KEY);
        log.info("{} size {}", MQ_STREAM_KEY, size);
        return size;
    }

    @Override
    public void createGroup(String groupName) {
        String str = stringRedisTemplate.opsForStream().createGroup(MQ_STREAM_KEY, groupName);
        log.info("create group :{}", str);
    }

    @Override
    public boolean destroyGroup(String groupName) {
        boolean b = stringRedisTemplate.opsForStream().destroyGroup(MQ_STREAM_KEY, groupName);
        log.info("destroy group :{}", b);
        return b;
    }

    /**
     * 当读取不到时，可以通过 SETID
     * XGROUP SETID mq:stream testcon 0 进行复位，这样就可以通过：
     * xreadgroup group testcon worker STREAMS mq:stream > 找到元素了。
     * （当无法通过 > 找元素时，可能是由于木有进行复位的关系）
     *
     * @param groupName
     * @param consumerName
     * @return
     */
    @Override
    public Set<String> readGroup(String groupName, String consumerName) {
        Set<String> set = new HashSet<>();
        //组名 + 消费者名
        Consumer consumer = Consumer.from(groupName, consumerName);
        //XREADGROUP 
        List<MapRecord<String, Object, Object>> messages = stringRedisTemplate.opsForStream().read(consumer, StreamOffset.create(MQ_STREAM_KEY, ReadOffset.lastConsumed()));
        for (MapRecord<String, Object, Object> message : messages) {
            log.info("RecordId:{},Stream:{},Value:{}", message.getId(), message.getStream(), message.getValue());
            set.add(message.getId() + ":" + message.getValue());
        }
        return set;
    }

    @Override
    public boolean ack(String groupName, String recordId) {
        long num = stringRedisTemplate.opsForStream().acknowledge(MQ_STREAM_KEY, groupName, RecordId.of(recordId));
        log.info("ack num:{}", num);
        return num > 0;
    }


}
